package org.fhm.zdte.client.service.proxy;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.fhm.substrate.annotation.Component;
import org.fhm.substrate.annotation.Setup;
import org.fhm.substrate.service.LoggerHandler;
import org.fhm.substrate.standard.ILogger;
import org.fhm.zdte.client.pojo.ProxyRequestBody;
import org.fhm.zdte.client.standard.INotification;
import org.fhm.zdte.client.standard.IObserver;
import org.fhm.zdte.client.standard.IProxyRequestSender;
import org.fhm.zdte.common.constant.DataTransform;
import org.fhm.zdte.common.exception.ProtocolException;
import org.fhm.zdte.common.pojo.Result;
import org.fhm.zdte.common.standard.IJsonConvertor;
import org.fhm.zdte.common.util.CommonUtil;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Stream;


/**
 * Customized proxy handler of netty.
 *
 * @author 谭波
 * @since 2024/1/12
 */
@SuppressWarnings("ALL")
@Component
@ChannelHandler.Sharable
public class ProxyRequestHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> implements IObserver {

    private final ILogger logger = LoggerHandler.getLogger(ProxyRequestHandler.class);

    private final Map<String, ChannelHandlerContext> channelHandlerContextMap = new ConcurrentHashMap<>();

    private final Map<String, String[]> cacheStringContainer = new ConcurrentHashMap<>();

    @Setup
    private IJsonConvertor iJsonConvertor;

    @Setup
    private CommonUtil commonUtil;

    @Setup
    private IProxyRequestSender iProxyRequestSender;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        channelHandlerContextMap.put(obtainChannelLongId(ctx), ctx);
        super.channelActive(ctx);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ch, TextWebSocketFrame msg) {
        ProxyRequestBody body = isReceiveComplete(msg);
        if (Objects.nonNull(body)) {
            String response;
            try {
                response = iProxyRequestSender.request(body);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
                Result<Void> result = Result.getFailMessage(e.getMessage());
                result.setMessageNumber(body.getMessageNumber());
                response = iJsonConvertor.toJson(result);
            }
            ch.writeAndFlush(new TextWebSocketFrame(response));
        }
    }

    /**
     * Accept the data and verify that it is complete.
     *
     * @param msg {@link TextWebSocketFrame}
     * @return {@link ProxyRequestBody}
     */
    private ProxyRequestBody isReceiveComplete(TextWebSocketFrame msg) {
        String text = msg.text();
        if (Objects.isNull(text) || text.isEmpty())
            throw new ProtocolException("there was an error in the broker message");
        String[] split = text.split(DataTransform.PROXY_PACKET_HEAD_SEPARATOR);
        int messageLength = Integer.parseInt(split[0]);
        int messageSerial = Integer.parseInt(split[1]);
        String messageNumber = split[2];
        String message = split[3];
        String[] strings;
        if (Objects.nonNull((strings = cacheStringContainer.get(messageNumber)))) {
            synchronized (strings) {
                if (Objects.nonNull((strings = cacheStringContainer.get(messageNumber))))
                    strings[messageSerial] = message;
            }
        } else {
            if (messageLength < 0)
                throw new ProtocolException("the transfer data exceeds the upper limit");
            strings = new String[messageLength];
            cacheStringContainer.put(messageNumber, strings);
            strings[messageSerial] = message;
        }
        if (Stream.of(strings).allMatch(Objects::nonNull)) {
            StringBuilder result = new StringBuilder();
            for (String string : strings)
                result.append(string);
            cacheStringContainer.remove(messageNumber);
            String[] finalResultSplit = result.toString().split(DataTransform.PROXY_PACKET_SEPARATOR);
            ProxyRequestBody body = iJsonConvertor.fromJson(finalResultSplit[0], ProxyRequestBody.class);
            if (finalResultSplit.length == 2)
                body.getParameters().put(DataTransform.DATA, finalResultSplit[1]);
            return body;
        }
        return null;
    }


    @Override
    public void receive(INotification iNotification) {
        Iterator<ChannelHandlerContext> iterator = channelHandlerContextMap.values().iterator();
        while (iterator.hasNext()) {
            ChannelHandlerContext chx = iterator.next();
            if (isDisconnect(chx))
                iterator.remove();
            else
                chx.writeAndFlush(new TextWebSocketFrame("NOTIFICATION:" + iJsonConvertor.toJson(iNotification)));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof IOException) {
            ctx.channel().close();
            logger.warn("a client has closed the connection abruptly: {}", cause.getMessage());
            channelHandlerContextMap.remove(obtainChannelLongId(ctx));
        } else if (isDisconnect(ctx)) {
            channelHandlerContextMap.remove(obtainChannelLongId(ctx));
            super.exceptionCaught(ctx, cause);
        } else
            super.exceptionCaught(ctx, cause);
    }

    private String obtainChannelLongId(ChannelHandlerContext ctx) {
        return ctx.channel().id().asLongText();
    }

    private boolean isDisconnect(ChannelHandlerContext chx) {
        Channel channel;
        return chx.isRemoved() || !(channel = chx.channel()).isActive() || !channel.isOpen();
    }
}
